package com.shujia.base;

import com.shujia.utils.HbaseUtil;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.*;

/**
 * 查询文科一班学生总分排名前10的学生（输出：学号，姓名，班级，总分）结果写到hbase
 */
public class HbaseTest {
    public static void main(String[] args) throws Exception {
        /**
         * 将学生成绩数据添加到hbase表中
         */
//        putScoreData();

        /**
         * 读取成绩表，计算每个人的总分，写入到hbase另一张表中
         */
//        HbaseUtil.truncateTable("sumScore");
//        getSumScoreStudents();

        /**
         * 获取前几的学生数据 学号，姓名，班级，总分
         */
        getTopNStudents();


    }

    public static void getTopNStudents() {
        try {
            Scanner sc = new Scanner(System.in);
            System.out.print("请输入要查询的前topN个数：");
            int topN = sc.nextInt();
            //TODO:学号，姓名，班级，总分

            //获取要查询的表
            Table sumScore = HbaseUtil.getOneTable("sumScore");
            Table students = HbaseUtil.getOneTable("students");

            //创建sumScore表的Scan对象,并设置limit的个数
            Scan scoreScan = new Scan();
            scoreScan.setLimit(topN);
            //创建集合存储查询的结果
            TreeMap<String, String> sumScoreMap = new TreeMap<>(new Comparator<String>() {
                @Override
                public int compare(String o1, String o2) {
                    //显示条件
                    return (o2.compareTo(o1) == 0) ? -1 : o2.compareTo(o1);
                }
            });
            //查询数据得到结果
            ResultScanner resultScanner = sumScore.getScanner(scoreScan);
            for (Result result : resultScanner) {
                String id = Bytes.toString(result.getRow()).split("_")[1];
                String score = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("sum_score")));
                sumScoreMap.put(score, id);
            }


            //拿着第一张表查询的结果查询学生的详细信息
            //使用获取所有键值对的方式遍历map集合
//            Set<Map.Entry<String, String>> entries = sumScoreMap.entrySet();
//            for (Map.Entry<String, String> entry : entries) {
//                String id = entry.getKey();
//                String score = entry.getValue();
//            }
            //使用获取所有键，然后根据键获取值的方式遍历map集合
            //使用lambda表达式，消费型接口遍历map集合

            sumScoreMap.forEach((k, v) -> {
                try {
                    Get get = new Get(Bytes.toBytes(v));
                    Result result = students.get(get);
                    //获取姓名
                    String name = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name")));
                    String clazz = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("clazz")));
                    System.out.println("学号：" + v + ", 姓名：" + name + ", 班级：" + clazz + ", 总分：" + k);
                    System.out.println("-----------------------------------------------------[");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void getSumScoreStudents() throws Exception {
        Table scores = HbaseUtil.getOneTable("scores");

        Scan scan = new Scan();
        ResultScanner resultScanner = scores.getScanner(scan);

        //创建Map集合存储学号和总分的关系
        HashMap<String, Integer> scoreMap = new HashMap<>();

        //遍历每一行数据
        for (Result result : resultScanner) {
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                //获取行键
                String id = Bytes.toString(CellUtil.cloneRow(cell)).split("-")[0];
                int score = Integer.parseInt(Bytes.toString(CellUtil.cloneValue(cell)));
                if (!scoreMap.containsKey(id)) {
                    scoreMap.put(id, score);
                } else {
                    scoreMap.put(id, scoreMap.get(id) + score);
                }
            }
        }
//        scoreMap.forEach((k, v) -> System.out.println("学号: " + k + ", 总分: " + v));

        HbaseUtil.createOneTable("sumScore", "info");

        scoreMap.forEach((k, v) -> HbaseUtil.putOneColData("sumScore", (1000 - v) + "_" + k, "info", "sum_score", v.toString()));


    }

    public static void putScoreData() throws Exception {

        HbaseUtil.createOneTable("scores", "info");

        BufferedReader br = new BufferedReader(new FileReader("hbase/data/score.txt"));
        String line = null;
        while ((line = br.readLine()) != null) {
            String[] infos = line.split(",");
            /**
             * hbase中行键的设计原则：
             * 1、唯一性
             * 2、散列性
             * 3、长度不宜过长
             */
            HbaseUtil.putOneColData("scores", infos[0] + "-" + infos[1], "info", "subject_score", infos[2]);
        }
    }
}
